//package com.open.source.demo;
//
//import com.open.source.MqConstant;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.apache.kafka.common.header.Headers;
//import org.apache.kafka.common.header.internals.RecordHeader;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.annotation.RetryableTopic;
//import org.springframework.kafka.retrytopic.FixedDelayStrategy;
//import org.springframework.messaging.handler.annotation.Header;
//import org.springframework.retry.annotation.Backoff;
//import org.springframework.stereotype.Component;
//
///**
// * @author ZonLen since on 2022/8/17 下午1:15
// */
//@Slf4j
//@Component
//public class DemoKafkaMethodAnnotationListener {
//
//
////  @KafkaListener(topics = MqConstant.TAX_TOPIC_DEMO, groupId = "YONG-YOU-TAX")
////  public void handler(ConsumerRecord<String, String> record, @Header("host-ip") String hostIp,
////      @Header("trance-id") String tranceId,
////      @Header("sender-model") String senderModel) {
////    log.info(
////        "@KafkaListener: host-ip={}, trance-id={}, sender-model＝{}, topic={}, partition={}, offset ＝{}, key ＝{}, value ＝{}",
////        hostIp, tranceId, senderModel, record.topic(),
////        record.partition(), record.offset(), record.key(), record.value());
////  }
//
//  @RetryableTopic(backoff = @Backoff(delay = 3000)
//      ,fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC
//  )
//  @KafkaListener(topics = MqConstant.TAX_TOPIC_DELAY_DEMO
//      , properties = "value.deserializer.encoding=GBK"
//  )
//  public void handlerDlt(ConsumerRecord<String, String> record, @Header(MqConstant.PRODUCER_HOST_IP) String hostIp,
//      @Header("trance-id") String tranceId,
//      @Header(MqConstant.PRODUCER_APPLICATION_NAME) String senderModel) {
//    final Headers headers = record.headers();
//    final String timeStr = new String(headers.lastHeader(MqConstant.X_RETRIES_HEADER).value());
//    log.info("@RetryableTopic: time ={}, topic={}, offset ＝{}, val={}",
//        timeStr, record.topic(), record.offset(), record.value());
//    final int time = Integer.parseInt(timeStr);
//    final int curTime = time + 1;
//    headers.remove(MqConstant.X_RETRIES_HEADER)
//        .add(new RecordHeader(MqConstant.X_RETRIES_HEADER, String.valueOf(curTime).getBytes()));
//  }
//
//}
